package org.niit.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

public class KafkaProducerParameters {


    public static void main(String[] args) throws ExecutionException, InterruptedException {

        //1.配置信息 保存在Properties
        Properties props = new Properties();
        //连接地址
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092");
        props.put(ProducerConfig.ACKS_CONFIG,"all");// 应答机制   0:效率最高，安全性最低    1：效率中等，安全性中等   -1(all):效率最低，安全性最大 和钱相关使用-1
        //提高Kafka生产者吞吐量  提高生产效率要加的配置
        //设置缓冲区的大小  单位是 B
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG,32*1024*1024);
        //设置批次大小 单位 是 B
        props.put(ProducerConfig.BATCH_SIZE_CONFIG,16*1024*1024);
        //设置等待时间
        props.put(ProducerConfig.LINGER_MS_CONFIG,1000);
        //设置发送失败的重试次数
        props.put(ProducerConfig.RETRIES_CONFIG,3);
        //设置压缩
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"snappy");
        //生产数据有序，如果没有开启幂等性设为1 开启后可以不写该参数
        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION,1);

        //序列号数据  我们生产的数据是字符串，Kafka所有的接收的数据是二进制，所有要字符串进行序列化
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());

        //2.创建生产者对象
        KafkaProducer<String,String> producer = new KafkaProducer<>(props);

        //3.发送数据
        for (int  i = 1; i<=100;i++){
            //构建一条数据
            ProducerRecord<String,String> record = new ProducerRecord<>("BD2",null,i+"");

            Future<RecordMetadata> res = producer.send(record);
            //发送成功的应答
            RecordMetadata metadata = res.get();//获得发送成功的元素 包含 主题 分区 偏移量 Key value 时间等数据
            System.out.println("第"+i+"条数据发送成功"+metadata.timestamp());

        }
        //4.关闭
        producer.close();

    }

}
